package controller

import (
	kc "common/kafkacluster/consumer"
	"iot-bridge/iot_cmd_up/ping/service"

	log "common/log4go.v1"
)

type Work struct {
	consumer      *kc.Consumer
	seqNo         int
	queueName     string
	hsetName      string
	monitor       chan int
	finishOneTask bool //是否完成一次完整生产
	dataServ      *service.PingUpServ
}

func (this *Work) StartWork(factoryWillClose *bool) {
	log.Info("worker(%d) begin start work", this.seqNo)
	var needRestart bool
	for {
		this.finishOneTask = true
		if *factoryWillClose {
			break
		}

		m, ok := <-this.consumer.Messages()
		if !ok {
			log.Info("kafka channel of worder(%d) close", this.seqNo)
			//if factory close  then quit ; if kafka err  restart
			if *factoryWillClose {
				continue
			} else {
				needRestart = true
				break
			}
		}
		this.finishOneTask = false

		//insert to database
		restartFlag, err := this.dataServ.Push(m.Value)
		if err != nil {
			log.Error("deal failed worker(%d) value(%s) error(%v)", this.seqNo, string(m.Value), err)
			if restartFlag {
				needRestart = true
				break
			} else {
				this.consumer.CommitUpto(m)
			}
		} else {
			log.Info("deal success worker(%d) value(%s)", this.seqNo, string(m.Value))
			this.consumer.CommitUpto(m)
		}
	}

	if needRestart {
		this.Restart()
	}

}

func (this *Work) Stop() {
	this.finishOneTask = true
	log.Info("worker(%d) begin to stop", this.seqNo)
	if this.consumer != nil {
		if err := this.consumer.Close(); err != nil {
			log.Error("this.consumer.Close() error(%v)", err)
		}
		this.consumer = nil
	}

	log.Info("worker(%d) stop  ok ", this.seqNo)

}

func (this *Work) Restart() {
	this.finishOneTask = true
	log.Info("worker(%d) begin to restart", this.seqNo)
	if this.consumer != nil {
		if err := this.consumer.Close(); err != nil {
			log.Error("this.consumer.Close() error(%v)", err)
		}
		this.consumer = nil
	}
	log.Info("worker(%d)  has stoped  for restart", this.seqNo)
	this.monitor <- this.seqNo
}
